package com.bw.gmall.realtime.app.dwd;

import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdToolCouponGet {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // TODO 3. 从 Kafka 读取业务数据
        tenv.executeSql("create table `topic_db` (" +
                "`database` string, " +
                "`table` string, " +
                "data map<string,string>, " +
                "`type` string, " +
                "`ts` string " +
                ")" + MyKafkaUtil.getKafkaDDL("topic_db","dwd_tool_coupon_get"));

        // TODO 4. 读取优惠券领用数据，封装为表
        Table resTable = tenv.sqlQuery("select " +
                "data['id'], " +
                "data['coupon_id'], " +
                "data['user_id'], " +
                "date_format(data['get_time'],'yyyy-MM-dd') date_id, " +
                "data['get_time'], " +
                "ts " +
                "from topic_db " +
                "where `table` = 'coupon_use'" +
                "and `type` = 'insert' ");

        tenv.createTemporaryView("res_table",resTable);


        // TODO 5. 建立 Kafka-Connector dwd_tool_coupon_get 表

        tenv.executeSql("create table dwd_tool_coupon_get(" +
                "id string, " +
                "coupon_id string, " +
                "user_id string, " +
                "date_id string, " +
                "get_time string, " +
                "ts string " +
                ")" + MyKafkaUtil.getKafkaSinkDDL("dwd_tool_coupon_get"));

        tenv.executeSql("insert into dwd_tool_coupon_get select * from res_table");


    }
}
